 //package net.bwie.realtime.gd.utils2;
//
//import com.alibaba.fastjson.JSONObject;
//import com.google.common.collect.Sets;
//import org.apache.commons.lang3.StringUtils;
//import org.apache.flink.configuration.Configuration;
//import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
//
//import java.sql.Connection;
//import java.sql.DriverManager;
//import java.sql.PreparedStatement;
//import java.sql.SQLException;
//import java.util.List;
//import java.util.Set;
//
//
//public class DimSinkFunction extends RichSinkFunction<JSONObject> {
//    public static List<String> cloumnList = ListUtil.toList("key", "value", "values", "like", "table", "view", "date", "datetime");
//    private Connection connection;
//
//    private String schema;
//    private String info;
//
//    public DimSinkFunction() {
//
//    }
//
//    public DimSinkFunction(String schema) {
//        this.schema = schema;
//    }
//
//    public DimSinkFunction(String schema, String info) {
//        this.schema = schema;
//        this.info = info;
//    }
//
//    @Override
//    public void open(Configuration parameters) throws Exception {
//        Class.forName(GmallConfig.PHOENIX_DRIVER);
//        connection = DriverManager.getConnection(GmallConfig.PHOENIX_URL);
//        connection.setAutoCommit(true);
//    }
//
//    @Override
//    public void invoke(JSONObject value, Context context) throws Exception {
//
//        PreparedStatement preparedStatement = null;
//        try {
//            String sinkTable = value.getJSONObject("source").getString("table");
//            sinkTable = "dim_" + sinkTable;
//            JSONObject data = value.getJSONObject("after");
//            Set<String> strings = Sets.newHashSet(data.keySet());
//            strings.remove("id");
//            String sinkColumns = StringUtils.join(strings, ",");
//            buildSchema();
//            buildTable(sinkTable, sinkColumns, null, null);
//            String upsertSql = genUpsertSql(sinkTable,
//                    data);
//            System.out.println(upsertSql);
//            preparedStatement = connection.prepareStatement(upsertSql);
//            preparedStatement.executeUpdate();
//
//        } catch (SQLException e) {
//            e.printStackTrace();
//        } finally {
//            if (preparedStatement != null) {
//                preparedStatement.close();
//            }
//        }
//
//    }
//
//    private void buildSchema() {
//        PreparedStatement preparedStatement = null;
//        try {
//            preparedStatement = connection.prepareStatement("create schema if not exists " + schema);
//            preparedStatement.execute();
//        } catch (SQLException e) {
//            throw new RuntimeException("Phoenix database " + schema + " create fail！");
//        } finally {
//            if (preparedStatement != null) {
//                try {
//                    preparedStatement.close();
//                } catch (SQLException e) {
//                    e.printStackTrace();
//                }
//            }
//        }
//    }
//
//    private String genUpsertSql(String sinkTable, JSONObject data) {
//        Object id = data.get("id");
//        data.remove("id");
//        JSONObject newData = new JSONObject();
//        data.forEach((k, v) -> {
//            if (v == null) {
//                v = "";
//            }
//            if (StrUtil.isBlank(info)) {
//                newData.put(wrapperField(k), v.toString().replace("'", ""));
//            } else {
//                newData.put(info + "." + wrapperField(k), v.toString().replace("'", ""));
//            }
//        });
//
//        return "upsert into " + schema + "." + sinkTable + "(id," +
//                StringUtils.join(newData.keySet(), ",") + ") values(" + id + ", '" +
//                StringUtils.join(newData.values(), "','") + "')";
//    }
//
//    private void buildTable(String sinkTable, String sinkColumns, String sinkPk, String sinkExtend) {
//
//        PreparedStatement preparedStatement = null;
//
//        try {
//            if (sinkPk == null) {
//                sinkPk = "id";
//            }
//            if (sinkExtend == null) {
//                sinkExtend = "";
//            }
//
//            StringBuffer createTableSQL = new StringBuffer("create table if not exists ")
//                    .append(schema)
//                    .append(".")
//                    .append(sinkTable)
//                    .append("( id bigint primary key ,");
//
//            String[] fields = sinkColumns.split(",");
//            for (int i = 0; i < fields.length; i++) {
//                String field;
//                if (StrUtil.isBlank(info)) {
//                    field = wrapperField(fields[i]);
//                } else {
//                    field = info + "." + wrapperField(fields[i]);
//                }
//
//                createTableSQL.append(field).append(" varchar ");
//
//                if (i < fields.length - 1) {
//                    createTableSQL.append(",");
//                }
//            }
//
//            createTableSQL.append(")").append(sinkExtend);
//
//            System.out.println(createTableSQL);
//
//            preparedStatement = connection.prepareStatement(createTableSQL.toString());
//            preparedStatement.execute();
//        } catch (SQLException e) {
//            throw new RuntimeException("Phoenix table" + sinkTable + " create fail ！");
//        } finally {
//            if (preparedStatement != null) {
//                try {
//                    preparedStatement.close();
//                } catch (SQLException e) {
//                    e.printStackTrace();
//                }
//            }
//        }
//    }
//
//
//    String wrapperField(String o) {
//
//        if (cloumnList.contains(o.toLowerCase())) {
//            o += "_v2";
//        }
//        return o;
////        return CaseFormat.LOWER_UNDERSCORE
////                .to(CaseFormat.LOWER_CAMEL, o.toLowerCase());
//    }
//}
